package com.study.spark.scala.cache_persist_checkpoint

import org.apache.spark.storage.StorageLevel
import org.apache.spark.{SparkConf, SparkContext}

/**
 *
 * @author stephen
 * @date 2019-09-27 11:36
 */
class StorageDemo {

  def main(args: Array[String]): Unit = {
    val conf = new SparkConf()
      .setMaster("local[*]")
      .setAppName("StorageDemo")
    val sc = new SparkContext(conf)

    val rdd = sc.makeRDD(Seq(1, 2, 3, 4, 5, 6, 7, 8, 9))

    // cache 底层调用的是 persist 方法，存储等级为: memory only
    rdd.cache()
    // cache和persist的区别：
    // cache只有一个默认的缓存级别MEMORY_ONLY，
    // 而persist可以根据情况设置其它的缓存级别。
    rdd.persist(StorageLevel.MEMORY_AND_DISK)

    // checkpoint将RDD持久化到文件系统中
    // checkpoint与persist的区别:
    // checkpoint会切断此RDD之前的依赖关系，
    // 而persist会保留依赖关系(persist会存在缓存失效的问题)。
    sc.setCheckpointDir("./ckp")
    rdd.cache() //避免两次计算
    rdd.checkpoint()

    //checkpoint的两大作用：
    // 一是spark程序长期驻留，过长的依赖会占用很多的系统资源，定期checkpoint可以有效的节省资源；
    // 二是维护过长的依赖关系可能会出现问题，一旦spark程序运行失败，RDD的容错成本会很高。
    // 注意：checkpoint执行前要先进行cache，避免两次计算。
    // 两次计算指的是：第一次是在程序运行的时候，第二次则是 Checkpoint 的时候就需要把这个 RDD 的转换关系重新计算一次。
    // 参考:https://blog.csdn.net/learn_tech/article/details/86511000
  }
}
